//! 集群命令处理模块
//! 
//! 实现Redis集群相关的所有命令，包括：
//! - 节点管理命令 (CLUSTER MEET, CLUSTER FORGET等)
//! - 槽位管理命令 (CLUSTER ADDSLOTS, CLUSTER DELSLOTS等)
//! - 状态查询命令 (CLUSTER NODES, CLUSTER INFO等)
//! - 数据迁移命令 (CLUSTER MIGRATE, CLUSTER SETSLOT等)

use crate::cluster::traits::{
    CommandProcessor, ClusterCommand, CommandResult as CommandResultTrait, CommandStats
};
use crate::cluster::{ClusterCoordinator, ClusterError, ClusterResult};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::info;

/// 集群命令解析器和处理器
pub struct ClusterCommandProcessor {
    /// 集群协调器引用
    coordinator: Arc<RwLock<ClusterCoordinator>>,
    
    /// 命令统计信息
    stats: Arc<RwLock<ClusterCommandStats>>,
}

/// 命令统计信息
#[derive(Debug, Clone, Default)]
pub struct ClusterCommandStats {
    pub total_commands: u64,
    pub successful_commands: u64,
    pub failed_commands: u64,
    pub commands_by_type: HashMap<String, u64>,
    pub total_processing_time: Duration,
    pub start_time: Option<Instant>,
}

impl From<ClusterCommandStats> for CommandStats {
    fn from(stats: ClusterCommandStats) -> Self {
        let avg_time = if stats.total_commands > 0 {
            stats.total_processing_time / stats.total_commands as u32
        } else {
            Duration::from_secs(0)
        };
        
        CommandStats {
            total_commands: stats.total_commands,
            successful_commands: stats.successful_commands,
            failed_commands: stats.failed_commands,
            commands_by_type: stats.commands_by_type,
            average_processing_time: avg_time,
        }
    }
}

/// Redis集群命令枚举
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RedisClusterCommand {
    // 节点管理命令
    Meet { ip: String, port: u16 },
    Forget { node_id: String },
    Replicate { node_id: String },
    
    // 槽位管理命令
    AddSlots { slots: Vec<u16> },
    DelSlots { slots: Vec<u16> },
    FlushSlots,
    SetSlot { slot: u16, subcommand: String, node_id: Option<String> },
    
    // 故障恢复命令
    Failover { option: Option<String> },
    
    // 配置命令
    Reset { option: Option<String> },
    SaveConfig,
    
    // 信息查询命令
    Nodes,
    Info,
    Slots,
    KeySlot { key: String },
    CountKeysInSlot { slot: u16 },
    GetKeysInSlot { slot: u16, count: u32 },
    
    // 数据迁移命令
    SetSlotStable { slot: u16 },
    SetSlotMigrating { slot: u16, node_id: String },
    SetSlotImporting { slot: u16, node_id: String },
    
    // 调试和维护命令
    MyId,
    Bumpepoch,
}

impl ClusterCommand for RedisClusterCommand {
    fn command_type(&self) -> &str {
        match self {
            Self::Meet { .. } => "MEET",
            Self::Forget { .. } => "FORGET",
            Self::Replicate { .. } => "REPLICATE",
            Self::AddSlots { .. } => "ADDSLOTS",
            Self::DelSlots { .. } => "DELSLOTS",
            Self::FlushSlots => "FLUSHSLOTS",
            Self::SetSlot { .. } => "SETSLOT",
            Self::Failover { .. } => "FAILOVER",
            Self::Reset { .. } => "RESET",
            Self::SaveConfig => "SAVECONFIG",
            Self::Nodes => "NODES",
            Self::Info => "INFO",
            Self::Slots => "SLOTS",
            Self::KeySlot { .. } => "KEYSLOT",
            Self::CountKeysInSlot { .. } => "COUNTKEYSINSLOT",
            Self::GetKeysInSlot { .. } => "GETKEYSINSLOT",
            Self::SetSlotStable { .. } => "SETSLOTSTABLE",
            Self::SetSlotMigrating { .. } => "SETSLOTMIGRATING",
            Self::SetSlotImporting { .. } => "SETSLOTIMPORTING",
            Self::MyId => "MYID",
            Self::Bumpepoch => "BUMPEPOCH",
        }
    }
    
    fn arguments(&self) -> &[String] {
        // 返回空切片，因为参数已经嵌入在枚举中
        &[]
    }
    
    fn validate(&self) -> ClusterResult<()> {
        match self {
            Self::Meet { ip, port } => {
                if ip.is_empty() {
                    return Err(ClusterError::Config("IP地址不能为空".to_string()));
                }
                if *port == 0 {
                    return Err(ClusterError::Config("端口号不能为0".to_string()));
                }
            }
            Self::Forget { node_id } | Self::Replicate { node_id } => {
                if node_id.is_empty() {
                    return Err(ClusterError::Config("节点ID不能为空".to_string()));
                }
            }
            Self::AddSlots { slots } | Self::DelSlots { slots } => {
                if slots.is_empty() {
                    return Err(ClusterError::Config("槽位列表不能为空".to_string()));
                }
                for &slot in slots {
                    if slot >= 16384 {
                        return Err(ClusterError::Config(format!("无效的槽位号: {slot}")));
                    }
                }
            }
            Self::SetSlot { slot, subcommand, .. } => {
                if *slot >= 16384 {
                    return Err(ClusterError::Config(format!("无效的槽位号: {slot}")));
                }
                if !["MIGRATING", "IMPORTING", "STABLE", "NODE"]
                    .contains(&subcommand.to_uppercase().as_str()) 
                {
                    return Err(ClusterError::Config(format!("无效的子命令: {subcommand}")));
                }
            }
            _ => {} // 其他命令不需要特殊验证
        }
        Ok(())
    }
    
    fn serialize(&self) -> ClusterResult<String> {
        serde_json::to_string(self)
            .map_err(|e| ClusterError::Config(format!("序列化命令失败: {e}")))
    }
    
    fn deserialize(data: &str) -> ClusterResult<Self> {
        serde_json::from_str(data)
            .map_err(|e| ClusterError::Config(format!("反序列化命令失败: {e}")))
    }
}

/// 命令执行结果
#[derive(Debug, Clone)]
pub enum ClusterCommandResult {
    /// 简单字符串回复
    SimpleString(String),
    /// 批量字符串回复
    BulkString(String),
    /// 整数回复
    Integer(i64),
    /// 数组回复
    Array(Vec<String>),
    /// 错误回复
    Error(String),
    /// 空回复
    Null,
}

impl CommandResultTrait for ClusterCommandResult {
    fn is_success(&self) -> bool {
        !matches!(self, Self::Error(_))
    }
    
    fn data(&self) -> Option<&str> {
        match self {
            Self::SimpleString(s) | Self::BulkString(s) => Some(s),
            Self::Integer(_) => None, // 整数不返回字符串数据
            Self::Array(_) => None,   // 数组不返回单个字符串
            Self::Error(_) => None,   // 错误不返回数据
            Self::Null => None,
        }
    }
    
    fn error(&self) -> Option<&str> {
        match self {
            Self::Error(e) => Some(e),
            _ => None,
        }
    }
    
    fn serialize(&self) -> ClusterResult<String> {
        match self {
            Self::SimpleString(s) => Ok(format!("+{s}")),
            Self::BulkString(s) => Ok(format!("${}\r\n{}", s.len(), s)),
            Self::Integer(i) => Ok(format!(":{i}")),
            Self::Array(arr) => {
                let mut result = format!("*{}\r\n", arr.len());
                for item in arr {
                    result.push_str(&format!("${}\r\n{}\r\n", item.len(), item));
                }
                Ok(result)
            }
            Self::Error(e) => Ok(format!("-ERR {e}")),
            Self::Null => Ok("$-1".to_string()),
        }
    }
}

impl ClusterCommandProcessor {
    /// 创建新的命令处理器
    pub fn new(coordinator: Arc<RwLock<ClusterCoordinator>>) -> Self {
        Self { 
            coordinator,
            stats: Arc::new(RwLock::new(ClusterCommandStats::default())),
        }
    }
    
    /// 解析Redis集群命令
    pub fn parse_command(args: &[String]) -> ClusterResult<RedisClusterCommand> {
        if args.is_empty() {
            return Err(ClusterError::Config("命令不能为空".to_string()));
        }
        
        match args[0].to_uppercase().as_str() {
            "MEET" => {
                if args.len() != 3 {
                    return Err(ClusterError::Config("CLUSTER MEET 需要 ip 和 port 参数".to_string()));
                }
                let ip = args[1].clone();
                let port = args[2].parse::<u16>()
                    .map_err(|_| ClusterError::Config("无效的端口号".to_string()))?;
                Ok(RedisClusterCommand::Meet { ip, port })
            }
            
            "FORGET" => {
                if args.len() != 2 {
                    return Err(ClusterError::Config("CLUSTER FORGET 需要 node_id 参数".to_string()));
                }
                Ok(RedisClusterCommand::Forget { node_id: args[1].clone() })
            }
            
            "NODES" => Ok(RedisClusterCommand::Nodes),
            "INFO" => Ok(RedisClusterCommand::Info),
            "MYID" => Ok(RedisClusterCommand::MyId),
            
            _ => Err(ClusterError::Config(format!("未知的集群命令: {}", args[0]))),
        }
    }
}

#[async_trait]
impl CommandProcessor for ClusterCommandProcessor {
    type Command = RedisClusterCommand;
    type CommandResult = ClusterCommandResult;
    
    async fn process_command(&mut self, command: Self::Command) -> Self::CommandResult {
        let start_time = Instant::now();
        let command_type = command.command_type().to_string();
        
        // 更新统计信息
        {
            let mut stats = self.stats.write().await;
            stats.total_commands += 1;
            *stats.commands_by_type.entry(command_type.clone()).or_insert(0) += 1;
            if stats.start_time.is_none() {
                stats.start_time = Some(start_time);
            }
        }
        
        let result = match command {
            RedisClusterCommand::Meet { ip, port } => {
                match self.coordinator.read().await.meet_node(&format!("{ip}:{port}")).await {
                    Ok(_) => ClusterCommandResult::SimpleString("OK".to_string()),
                    Err(e) => ClusterCommandResult::Error(e.to_string()),
                }
            }
            
            RedisClusterCommand::Forget { node_id } => {
                match self.coordinator.write().await.forget_node(&node_id).await {
                    Ok(_) => ClusterCommandResult::SimpleString("OK".to_string()),
                    Err(e) => ClusterCommandResult::Error(e.to_string()),
                }
            }
            
            RedisClusterCommand::Nodes => {
                match self.coordinator.read().await.get_cluster_nodes().await {
                    Ok(nodes_info) => ClusterCommandResult::BulkString(nodes_info),
                    Err(e) => ClusterCommandResult::Error(e.to_string()),
                }
            }
            
            RedisClusterCommand::Info => {
                match self.coordinator.read().await.get_cluster_info().await {
                    Ok(info) => ClusterCommandResult::BulkString(format!(
                        "cluster_state:ok\ncluster_slots_assigned:{}\ncluster_slots_ok:{}\ncluster_slots_pfail:0\ncluster_slots_fail:0\ncluster_known_nodes:{}\ncluster_size:{}\ncluster_current_epoch:{}\ncluster_my_epoch:{}",
                        info.slots_assigned,
                        info.slots_assigned,
                        info.node_count,
                        info.cluster_size,
                        1, // TODO: 从集群状态获取实际epoch
                        1  // TODO: 从集群状态获取实际epoch
                    )),
                    Err(e) => ClusterCommandResult::Error(e.to_string()),
                }
            }
            
            RedisClusterCommand::MyId => {
                let node_id = self.coordinator.read().await.get_local_node_id().await;
                ClusterCommandResult::BulkString(node_id)
            }
            
            _ => ClusterCommandResult::Error("命令未实现".to_string()),
        };
        
        // 更新统计信息
        {
            let mut stats = self.stats.write().await;
            let processing_time = start_time.elapsed();
            stats.total_processing_time += processing_time;
            
            if result.is_success() {
                stats.successful_commands += 1;
            } else {
                stats.failed_commands += 1;
            }
        }
        
        result
    }
    
    async fn process_commands_batch(&mut self, commands: Vec<Self::Command>) -> Vec<Self::CommandResult> {
        let mut results = Vec::new();
        for command in commands {
            results.push(self.process_command(command).await);
        }
        results
    }
    
    async fn validate_command(&self, command: &Self::Command) -> ClusterResult<()> {
        command.validate()
    }
    
    fn supported_commands(&self) -> Vec<String> {
        vec![
            "MEET".to_string(),
            "FORGET".to_string(),
            "REPLICATE".to_string(),
            "ADDSLOTS".to_string(),
            "DELSLOTS".to_string(),
            "FLUSHSLOTS".to_string(),
            "SETSLOT".to_string(),
            "FAILOVER".to_string(),
            "RESET".to_string(),
            "SAVECONFIG".to_string(),
            "NODES".to_string(),
            "INFO".to_string(),
            "SLOTS".to_string(),
            "KEYSLOT".to_string(),
            "COUNTKEYSINSLOT".to_string(),
            "GETKEYSINSLOT".to_string(),
            "MYID".to_string(),
            "BUMPEPOCH".to_string(),
        ]
    }
    
    async fn get_command_stats(&self) -> CommandStats {
        let stats = self.stats.read().await;
        stats.clone().into()
    }
    
    fn register_command_handler<F>(&mut self, command_type: String, _handler: F)
    where
        F: Fn(&Self::Command) -> ClusterResult<Self::CommandResult> + Send + Sync + 'static,
    {
        // TODO: 实现命令处理器注册功能
        info!("注册命令处理器: {}", command_type);
    }
}

